---
title: python 后端服务
date: 2023-10-15 22:10:43 
images: ['https://kodo.huixiangwuyou.com/blog/images/nodecrawl.png']
layout: PostBanner 
summary: 'python 后台服务流行的框架有:flask、 django、 fastapi。django和fastapi都是python的web框架，django是一个完整的web框架，而fastapi是一个轻量级的web框架，django的优点是功能强大，缺点是性能差，fastapi的优点是性能好，缺点是功能不够强大。'
tags: 
 - python
---


## django

[官网](https://www.djangoproject.com/)

### 安装
```bash
pip install Django
# 检查是否安装成功
django-admin --version

django-admin startproject myproject
cd myproject
# 运行开发服务器
python manage.py runserver

```
### 路由模块化
```python
# myproject/urls.py
from django.contrib import admin
from django.urls import path, include

urlpatterns = [
    path('admin/', admin.site.urls),
    path('', include('myapp.urls')),  # 包含 myapp 的 URL 配置
]

# api/urls.py
from django.urls import path
from . import views

urlpatterns = [
    path('hello/', views.hello),
]

# api/views.py
from django.http import HttpResponse

def hello(request):
    return HttpResponse("Hello, Django!")

```

### 消息队列

```python
pip install celery redis
# myproject/celery.py
import os
from celery import Celery

os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'myproject.settings')
app = Celery('myproject', broker='redis://localhost:6379/0')
app.config_from_object('django.conf:settings', namespace='CELERY')
app.autodiscover_tasks()
# api/tasks.py
from celery import shared_task

@shared_task
def send_email():
    print("Sending email...")

celery -A myproject worker --loglevel=info

```

### 数据库
```python 
pip install mysqlclient  # 或 pymysql
# myproject/settings.py
DATABASES = {
    'default': {
        'ENGINE': 'django.db.backends.mysql',
        'NAME': 'mydb',
        'USER': 'root',
        'PASSWORD': '123456',
        'HOST': 'localhost',
        'PORT': '3306',
    }
}

# api/utils/db.py
from django.db import connection

def query_data(sql):
    with connection.cursor() as cursor:
        cursor.execute(sql)
        rows = cursor.fetchall()
    return rows

#  使用 ORM 查询数据
from django.db import models

class User(models.Model):
    name = models.CharField(max_length=100)
    email = models.EmailField()

# 查询数据
users = User.objects.all()

```

<br/>

## fastapi
[官网](https://fastapi.tiangolo.com/zh/)

### 安装
```bash
pip install fastapi
pip install uvicorn


# 安装 pyproject.toml 核心依赖 
pip install -e .

# 安装核心 + 开发依赖
pip install -e ".[dev]"

# 安装核心 + 测试依赖
pip install -e ".[test]"
```
> 安装Poetry 管理项目，将项目依赖写入 pyproject.toml 文件中
```bash
pip install poetry
poetry init

poetry add fastapi
```

### 快速开始
```python
from typing import Union

from fastapi import FastAPI

app = FastAPI()


@app.get("/")
def read_root():
    return {"Hello": "World"}


@app.get("/items/{item_id}")
def read_item(item_id: int, q: Union[str, None] = None):
    return {"item_id": item_id, "q": q}
```
运行
```bash
uvicorn main:app --reload    
```

> 访问 URL_ADDRESS访问 http://127.0.0.1:8000/items/5?q=somequery 可以看到返回结果
> 访问 http://127.0.0.1:8000/docs 可以看到 swagger 文档


### 路由模块化
```python
project/
│
├── main.py
├── routers/
│   ├── __init__.py
│   └── items.py
└── models/
    ├── __init__.py
    └── items.py

# routers/items.py
from fastapi import APIRouter

# 创建一个 APIRouter 实例
router = APIRouter()

# 定义一个路由
@router.get("/items/")
async def read_items():
  return {"message": "This is a list of items."}

# 定义另一个路由
@router.get("/items/{item_id}")
async def read_item(item_id: int):
  return {"item_id": item_id, "message": f"Item {item_id} is selected."}

# user.py
from fastapi import APIRouter
# 给模块添加前缀和标签
router = APIRouter( 
  prefix="/user",
  tags=["User"]
)
@router.get("/deatails/")
async def user_items():
  return {"message": "This is a deatail of user."}

# routers/__init__.py
from .items import router as items_router  
from .user import router as user_router  

# main.py
from fastapi import FastAPI
from routers import items_router,user_router
"""
引入多个路由模块，然后在下面挂载
"""

# 创建 FastAPI 应用实例
app = FastAPI()

# 挂载路由模块
app.include_router(items_router)

if __name__ == "__main__":
  import uvicorn
  uvicorn.run(app, host="0.0.0.0", port=8000)

```
![路由模块化](https://kodo.huixiangwuyou.com/blog/images/python/30406200637.png)


### GET|POST|PUT|DELETE

FASTapi支持所有HTTP方法，包括GET、POST、PUT、DELETE等，使用装饰器来定义路由，例如

```python
from fastapi import FastAPI

app = FastAPI()

@app.get("/items/")
async def read_items():
    return [{"name": "Item 1"}, {"name": "Item 2"}]

@app.post("/items/")
async def create_item(item: Item):  # 请求体为create_item 函数参数
    return item

@app.put("/items/{item_id}")
async def update_item(item_id: int, item: Item):
    return {"item_id": item_id, "item": item}

@app.delete("/items/{item_id}")
async def delete_item(item_id: int):
    return {"item_id": item_id}
```

> 注意：请求体参数需要使用Pydantic模型来定义，以便自动进行数据验证和序列化。

> GET请求用于获取资源，请求参数方式有多种

```python
from fastapi import FastAPI
from fastapi import Query
app = FastAPI()

# 查询参数
@app.get("/items/")
async def read_items(page: int = 1, limit: int = 10, search: str = None):
    return {
        "page": page,
        "limit": limit,
        "search": search,
    }
# 访问 http://127.0.0.1:8000/items/?page=2&limit=20&search=apple
# 参数默认是可选的（如果未传参，使用默认值）

@app.get("/items/")
async def read_items(page: int = Query(..., gt=0), limit: int = Query(10, le=100)):
# Query(..., gt=0) 表示 page 是必传的，且必须 >0
# Query(10, le=100) 表示 limit 默认 10，且必须 <=100    
    

# 路径参数
@app.get("/items/{item_id}")
async def read_item(item_id: int, q: str = None):
# item_id 是路径参数，直接从 URL 中提取   q 是查询参数（可选）。    
# http://127.0.0.1:8000/items/42?q=test


#  混合使用路径参数 + 查询参数
@app.get("/users/{user_id}/orders")
async def get_user_orders(
    user_id: int,
    page: int = 1,
    limit: int = 10,
    status: str = "pending"
):

# http://127.0.0.1:8000/users/123/orders?page=2&limit=5&status=completed


# 使用 Pydantic 模型接收多个参数（推荐）
from pydantic import BaseModel
from fastapi import Depends

class Pagination(BaseModel):
    page: int = 1
    limit: int = 10

@app.get("/articles/")
async def get_articles(pagination: Pagination = Depends()):
# http://127.0.0.1:8000/articles/?page=3&limit=20


# 接收数组/列表参数
from typing import List
from fastapi import Query

@app.get("/items/")
async def read_items(tags: List[str] = Query(...)):
# http://127.0.0.1:8000/items/?tags=python&tags=fastapi&tags=web    

```
| 传参方式 | 适用场景 | 示例 |
| --- | --- | --- |
| 查询参数 | 可选参数、分页、搜索 | ?page=1&limit=10 |
| 路径参数 | RESTful 资源定位 | /items/42 |
| 混合传参 | 路径 + 查询参数 | /users/123/orders?status=completed |
| Pydantic 模型 | 结构化复杂参数 | Depends(Pagination) |
| 数组参数 | 多值查询 | ?tags=python&tags=fastapi |
|  |  |  |




### 数据库
```python
pip install sqlalchemy  pymysql

from fastapi import FastAPI, Depends
from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

# 数据库连接配置，包含连接池相关参数
SQLALCHEMY_DATABASE_URL = "mysql+mysqlconnector://username:password@localhost:3306/dbname"

# 创建数据库引擎，配置连接池
engine = create_engine(
    SQLALCHEMY_DATABASE_URL,
    pool_size=5,  # 连接池中的初始连接数
    max_overflow=10,  # 连接池允许的最大额外连接数
    pool_timeout=30,  # 从连接池获取连接的超时时间（秒）
    pool_recycle=3600  # 连接在池中保持活动的时间（秒），超时后会被回收
)

# 创建会话工厂
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

# 创建基类
Base = declarative_base()


# 定义数据模型
class Item(Base):
    __tablename__ = "items"

    id = Column(Integer, primary_key=True, index=True)
    name = Column(String(255), index=True)
    description = Column(String(255))


# 创建数据库表
Base.metadata.create_all(bind=engine)

# 创建 FastAPI 应用
app = FastAPI()


# 依赖项，用于获取数据库会话
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()


# 简单的 API 端点，用于创建一个新的 Item
@app.post("/items/")
def create_item(name: str, description: str, db: sessionmaker = Depends(get_db)):
    new_item = Item(name=name, description=description)
    db.add(new_item)
    db.commit()
    db.refresh(new_item)
    return new_item
```    

[gitee 仓库](https://gitee.com/miss-tree/blog-fastapi)
[个人项目 swagger](https://fast.huixiangwuyou.com/docs)


> 自增id推荐使用雪花算法 pip install pysnowflake 



### 常见模块

- pydantic: 数据验证和序列化
- sqlalchemy: ORM 框架，用于与数据库交互
- uvicorn: ASGI 服务器,用于运行 FastAPI 应用,保证应用执行错误代码时不会崩溃和在生产环境中的稳定性
- starlette: 基础的 ASGI 框架
- jinja2: 模板引擎
- python-multipart: 处理文件上传
- python-dotenv: 环境变量管理
- pyJwt: JWT 认证










